package net.bwie.realtime.jtp.dwd.log.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.utils.DorisUtil;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.math.RoundingMode;

public class TradeOrderDetail {
    public static void main(String[] args) throws Exception {
// 1-执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);

        // 2-数据源：消费 ODS 层 Kafka
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "ods_trade_order_detail");
//        kafkaStream.print("kafka");

        // 3-解析 JSON
        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaStream.map(s -> JSON.parseObject(s));

        // 4-数据清洗
        SingleOutputStreamOperator<String> cleanedStream = jsonStream.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, Context context, Collector<String> collector) {

                // 必填字段校验
                String[] requiredFields = new String[]{
                        "order_id" ,
                        "event_id" ,
                        "live_room_id" ,
                        "anchor_id" ,
                        "user_id" ,
                        "sku_id" ,
                        "order_time",
                        "order_cnt",
                        "order_amount" ,
                        "status"
                };
                for (String field : requiredFields) {
                    //检查 JSON 数据中是否包含所有必需的字段，并且这些字段的值都不为 null，
                    // 如果有任何一个必需字段缺失或值为 null，就会跳过当前这条数据（不继续处理
                    if (!jsonObject.containsKey(field) || jsonObject.get(field) == null) {
                        return; // 跳过不完整数据
                    }
                }
                // 收集清洗后的数据
                collector.collect(jsonObject.toJSONString());
            }
        });

        cleanedStream.print("cleanedStream=========>");

        // 5-写入 DWD Kafka 主题
        KafkaUtil.producerKafka(cleanedStream, "dwd_trade_order_detail");

        // 6-触发执行
        env.execute("DWD_LiveRoomEventLog_Job");
    }
}
